package com.jubaozan.framework.sentinel.stat.traffic;

import com.jubaozan.framework.sentinel.stat.traffic.repository.TrafficLogRepository;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;

/**
 * @author jiangjinwu
 * @version 1.0
 * @since 2025/7/23 18:50
 */

@Slf4j
public class TrafficLogSaver {
    @Resource
    private TrafficLogRepository trafficLogRepository;

    /**
     * 每1秒执行一次，但按数据的时间戳归属到正确的秒级区间
     * 核心：不依赖任务执行时间，而是根据数据自带的timestamp划分
     */
    @Scheduled(fixedRate = 1000*2, initialDelay = 1000)
    public void savePerSecond() {
        // 1. 获取当前缓存的所有数据
        List<TrafficLogSecondCache.TrafficLog> allLogs = TrafficLogSecondCache.takeAllAndClear();
        if (allLogs.isEmpty()) {
            return;
        }

        // 2. 按数据的秒级时间戳分组（确保数据归属正确的秒）
        // 时间戳取整到秒：timestamp / 1000 * 1000
        Map<Long, List<TrafficLogSecondCache.TrafficLog>> secondGroup = allLogs.stream()
            .collect(Collectors.groupingBy(log -> log.getTimestamp().getTime() / 1000 * 1000));

        // 3. 过滤掉未来时间的数据（可能因时钟偏差导致）
        long currentSecond = System.currentTimeMillis() / 1000 * 1000;
        Map<Long, List<TrafficLogSecondCache.TrafficLog>> validGroup = secondGroup.entrySet().stream()
            .filter(entry -> entry.getKey() <= currentSecond) // 只处理当前及过去的秒级数据
            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        // 4. 异步批量保存（非阻塞，避免任务耗时过长）
        CompletableFuture.runAsync(() -> {
            try {
                // 合并同一秒+IP+资源的数据（应对极端情况下的重复）
                List<TrafficLogSecondCache.TrafficLog> mergedLogs = mergeDuplicateLogs(validGroup);
                trafficLogRepository.batchSave(mergedLogs);
            } catch (Exception e) {
                // 保存失败时记录日志（可后续重试）
                System.err.println("批量保存流量数据失败：" + e.getMessage());
            }
        });
    }

    /**
     * 合并同一秒、同一IP、同一资源的重复数据（避免并发导致的重复缓存）
     */
    private List<TrafficLogSecondCache.TrafficLog> mergeDuplicateLogs(Map<Long, List<TrafficLogSecondCache.TrafficLog>> group) {
        List<TrafficLogSecondCache.TrafficLog> merged = new ArrayList<>();
        for (List<TrafficLogSecondCache.TrafficLog> logs : group.values()) {
            // 按IP+资源分组，合并指标
            Map<String, TrafficLogSecondCache.TrafficLog> mergeMap = new HashMap<>();
            for (TrafficLogSecondCache.TrafficLog log : logs) {
                String key = log.getIp() + "_" + log.getResource();
                mergeMap.compute(key, (k, existing) -> {
                    if (existing == null) {
                        return log; // 新建记录
                    } else {
                        // 合并计数和RT
                        existing.setTotalCount(existing.getTotalCount() + log.getTotalCount());
                        existing.setSuccessCount(existing.getSuccessCount() + log.getSuccessCount());
                        existing.setExceptionCount(existing.getExceptionCount() + log.getExceptionCount());
                        existing.setTotalRt(existing.getTotalRt() + log.getTotalRt());
                        return existing;
                    }
                });
            }
            merged.addAll(mergeMap.values());
        }
        return merged;
    }
}
